/**
 * Copyright (c) 2022 Presto Labs Pte. Ltd.
 * Author: Ju
 */

#include "coin2/base/influxdb/influxdb_client.h"

#include <glog/logging.h>

#include "coin2/base/base64.h"

namespace coin2::base::influxdb {


InfluxdbClient::InfluxdbClient(boost::asio::io_context* ioc,
                               std::shared_ptr<Scheduler> timer,
                               std::string host, std::string org, uint16_t port,
                               std::string db, std::string token,
                               uint64_t flush_interval, uint32_t max_queue_size)
    : io_ctx_(ioc),
      timer_(timer),
      host_(host),
      org_(org),
      port_(port),
      token_(token),
      flush_ms_(flush_interval),
      last_flush_time_(0),
      max_queue_size_(max_queue_size) {
  target_ = "/api/v2/write?&org=" + org_ + "&bucket=" + db + "&precision=ns";
}

InfluxdbClient::~InfluxdbClient() {
  for (auto client : http_clients_) {
    client.reset();
  }
}

void InfluxdbClient::Run() {
  timer_->AddPeriodicTask(std::bind(&InfluxdbClient::OnTimer, this), 3 * kOneSec,
                          "time_report");
}

void InfluxdbClient::SendData(std::string data) { OnEvent(data); }

void InfluxdbClient::SendData(LineProtocol data) {
  std::ostringstream os;
  assert(data.header != nullptr);
  os << *data.header;

  auto fields = data.buffer;
  if (fields->size()) {
    bool first = true;
    for (const auto& f : *fields) {
      if (first) {
        os << " " << f.first << "=" << f.second;
        first = false;
      } else {
        os << "," << f.first << "=" << f.second;
      }
    }
  }
  os << " " << data.ts;
  OnEvent(os.str());
}

void InfluxdbClient::OnEvent(std::string data) {
  while (items_.size() >= kMaxQueueSize) {
    items_.pop();
  }
  items_.push(data);
  if (items_.size() > 10) {
    Flush();
  }
}

void InfluxdbClient::OnTimer() {
  DeleteClients();
  Flush();
}

void InfluxdbClient::Flush() {
  auto now = impl::GetCurrentTimestamp();
  auto time_ms = now - last_flush_time_ / 1000000;
  if (now - last_flush_time_ < flush_ms_ && items_.size() < 10) {
    return;
  }

  LOG(INFO) << "[" << time_ms << "ms] Flush cached data. " << items_.size() << " lines pending...";

  last_flush_time_ = now;
  if (items_.size() == 0) return;
  FlushNum(10);
}

int InfluxdbClient::FlushNum(int n) {
  int cnt = 0;
  std::vector<char> buffer;
  bool first = true;

  while (!items_.empty() && n) {
    auto data = items_.front();
    items_.pop();
    if (first) {
      first = false;
    } else {
      buffer.push_back('\n');
    }
    std::copy(data.begin(), data.end(), std::back_inserter(buffer));
    cnt++;
    n--;
  }
  std::string res;
  for (auto x : buffer) {
    if (x != '\n') {
      res.push_back(x);
    } else {
      res.push_back(' ');
    }
  }
  // LOG(INFO) << "Try to write buffer to influxdb: " << res;

  std::string buffer_str(buffer.begin(), buffer.end());

  RequestType* req = new RequestType(http::verb::post, target_, kVersion);
  req->set("Authorization", "Token " + token_);
  req->set("Content-Type:", "text/plain");
  req->set(http::field::host, host_);
  req->set(http::field::user_agent, BOOST_BEAST_VERSION_STRING);
  req->keep_alive(true);
  req->body() = std::move(buffer_str);
  req->prepare_payload();
  auto client = TakeClient();
  client->AsyncSend(req);
  req_map_.emplace(client, req);
  return cnt;
}

std::shared_ptr<HttpsConnection> InfluxdbClient::TakeClient() {
  if (http_clients_.empty()) {
    auto client = std::make_shared<HttpsConnection>(host_, port_, "", this,
                                                    io_ctx_, nullptr, "");
    client->Start();
    return client;
  }
  auto it = http_clients_.begin();
  auto client = *it;
  http_clients_.erase(it);

  return client;
}

void InfluxdbClient::DeleteReq(std::shared_ptr<HttpsConnection> conn) {
  auto iter = req_map_.find(conn);
  if (iter != req_map_.end()) {
    delete iter->second;
    iter->second = nullptr;
    req_map_.erase(iter);
  }
}

void InfluxdbClient::OnConnected(std::weak_ptr<HttpsConnection> conn) {}

void InfluxdbClient::OnRequestSent(std::weak_ptr<HttpsConnection> conn, const RequestType* req) {
  auto conn_ptr = conn.lock();
  if (conn_ptr) {
    DeleteReq(std::move(conn_ptr));
  }
}

void InfluxdbClient::OnResponseReceived(
    std::weak_ptr<HttpsConnection> conn,
    const ResponseType* rep) {
  RevertClient(std::move(conn));
}

void InfluxdbClient::OnError(std::weak_ptr<HttpsConnection> conn, boost::system::error_code ec) {
  DeferDeleteClient(std::move(conn));
}

void InfluxdbClient::RevertClient(std::weak_ptr<HttpsConnection> client) {
  if (http_clients_.size() >= kMaxFreeHttpClient) {
    DeferDeleteClient(client);
    return;
  }
  auto conn = client.lock();
  if (conn == nullptr) {
    return;
  }
  http_clients_.insert(std::move(conn));
}

void InfluxdbClient::DeferDeleteClient(std::weak_ptr<HttpsConnection> client) {
  auto conn = client.lock();
  if (conn == nullptr) {
    return;
  }
  http_clients_.erase(conn);
  if (!defer_deleted_clients_.count(conn)) {
    conn->AsyncShutdown();
    defer_deleted_clients_.emplace(conn);
  }
}

void InfluxdbClient::DeleteClients() {
  auto iter = defer_deleted_clients_.begin();
  while (iter != defer_deleted_clients_.end()) {
    const auto& conn = (*iter);
    if (conn->in_sending()) {
      ++iter;
      continue;
    }
    DeleteReq(conn);
    iter = defer_deleted_clients_.erase(iter);
  }
}

}  // namespace coin2::base::influxdb
